/*-------------------------------------------------------------------------
 *
 * pgoutput.c
 *		逻辑复制输出插件
 *
 * Copyright (c) 2012-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *		  src/backend/replication/pgoutput/pgoutput.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/tupconvert.h"
#include "catalog/partition.h"
#include "catalog/pg_publication.h"
#include "catalog/pg_publication_rel.h"
#include "commands/defrem.h"
#include "executor/executor.h"
#include "fmgr.h"
#include "nodes/makefuncs.h"
#include "optimizer/optimizer.h"
#include "replication/logical.h"
#include "replication/logicalproto.h"
#include "replication/origin.h"
#include "replication/pgoutput.h"
#include "utils/builtins.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/syscache.h"
#include "utils/varlena.h"

PG_MODULE_MAGIC;

extern void _PG_output_plugin_init(OutputPluginCallbacks *fc_cb);

static void fc_pgoutput_startup(LogicalDecodingContext *fc_ctx,
							 OutputPluginOptions *fc_opt, bool fc_is_init);
static void fc_pgoutput_shutdown(LogicalDecodingContext *fc_ctx);
static void fc_pgoutput_begin_txn(LogicalDecodingContext *fc_ctx,
							   ReorderBufferTXN *fc_txn);
static void fc_pgoutput_commit_txn(LogicalDecodingContext *fc_ctx,
								ReorderBufferTXN *fc_txn, XLogRecPtr fc_commit_lsn);
static void fc_pgoutput_change(LogicalDecodingContext *fc_ctx,
							ReorderBufferTXN *fc_txn, Relation fc_rel,
							ReorderBufferChange *fc_change);
static void fc_pgoutput_truncate(LogicalDecodingContext *fc_ctx,
							  ReorderBufferTXN *fc_txn, int fc_nrelations, Relation fc_relations[],
							  ReorderBufferChange *fc_change);
static void fc_pgoutput_message(LogicalDecodingContext *fc_ctx,
							 ReorderBufferTXN *fc_txn, XLogRecPtr fc_message_lsn,
							 bool fc_transactional, const char *fc_prefix,
							 Size fc_sz, const char *fc_message);
static bool fc_pgoutput_origin_filter(LogicalDecodingContext *fc_ctx,
								   RepOriginId fc_origin_id);
static void fc_pgoutput_begin_prepare_txn(LogicalDecodingContext *fc_ctx,
									   ReorderBufferTXN *fc_txn);
static void fc_pgoutput_prepare_txn(LogicalDecodingContext *fc_ctx,
								 ReorderBufferTXN *fc_txn, XLogRecPtr fc_prepare_lsn);
static void fc_pgoutput_commit_prepared_txn(LogicalDecodingContext *fc_ctx,
										 ReorderBufferTXN *fc_txn, XLogRecPtr fc_commit_lsn);
static void fc_pgoutput_rollback_prepared_txn(LogicalDecodingContext *fc_ctx,
										   ReorderBufferTXN *fc_txn,
										   XLogRecPtr fc_prepare_end_lsn,
										   TimestampTz fc_prepare_time);
static void fc_pgoutput_stream_start(struct LogicalDecodingContext *fc_ctx,
								  ReorderBufferTXN *fc_txn);
static void fc_pgoutput_stream_stop(struct LogicalDecodingContext *fc_ctx,
								 ReorderBufferTXN *fc_txn);
static void fc_pgoutput_stream_abort(struct LogicalDecodingContext *fc_ctx,
								  ReorderBufferTXN *fc_txn,
								  XLogRecPtr fc_abort_lsn);
static void fc_pgoutput_stream_commit(struct LogicalDecodingContext *fc_ctx,
								   ReorderBufferTXN *fc_txn,
								   XLogRecPtr fc_commit_lsn);
static void fc_pgoutput_stream_prepare_txn(LogicalDecodingContext *fc_ctx,
										ReorderBufferTXN *fc_txn, XLogRecPtr fc_prepare_lsn);

static bool publications_valid;
static bool in_streaming;

/*
 * 发布数据的私有内存上下文，在启动pgoutput时创建于
 * PGOutputData->context，并在其父上下文通过专用MemoryContextCallback重置时设为NULL。
 */
static MemoryContext pubctx = NULL;

static List *fc_LoadPublications(List *fc_pubnames);
static void fc_publication_invalidation_cb(Datum fc_arg, int fc_cacheid,
										uint32 fc_hashvalue);
static void fc_send_relation_and_attrs(Relation fc_relation, TransactionId fc_xid,
									LogicalDecodingContext *fc_ctx,
									Bitmapset *fc_columns);
static void fc_send_repl_origin(LogicalDecodingContext *fc_ctx,
							 RepOriginId fc_origin_id, XLogRecPtr fc_origin_lsn,
							 bool fc_send_origin);
static void fc_update_replication_progress(LogicalDecodingContext *fc_ctx,
										bool fc_skipped_xact);

/*
 * 仅使用3个发布操作进行行过滤（“插入”、“更新”、“删除”）。请参见RelationSyncEntry.exprstate[]。
 */
enum RowFilterPubAction
{
	PUBACTION_INSERT,
	PUBACTION_UPDATE,
	PUBACTION_DELETE
};

#define NUM_ROWFILTER_PUBACTIONS (PUBACTION_DELETE+1)

/*
 * 用于记住我们已发送的关系模式的映射中的条目。
 *
 * schema_sent标志决定当前关系的模式记录（以及如果设置publish_as_relid则为其祖先）是否已发送给订阅者（在这种情况下我们不需要再次发送）。
 *
 * 然而，下游的模式缓存仅在提交时更新，对于流式事务，提交顺序可能与发送事务的顺序不同。此外，（子）事务可能被中止，因此我们需要为每个（子）事务发送模式，以便在中止时不会丢失模式信息。为此，我们维护已发送模式的xids列表（streamed_txns）。
 *
 * 对于分区，'pubactions'不仅考虑表自身的发布，还考虑其所有祖先的发布。
 */
typedef struct RelationSyncEntry
{
	Oid			relid;			/* 关系oid */

	bool		replicate_valid;	/* 条目的整体有效性标志 */

	bool		schema_sent;
	List	   *streamed_txns;	/* 使用此模式的流式顶层事务 */

	/* 我们在发布这个关系吗？ */
	PublicationActions pubactions;

	/*
	 * 用于行过滤的ExprState数组。不同的发布操作不允许将多个表达式始终组合为一个，
	 * 因为更新或删除限制表达式中的列为复制身份索引的一部分，而插入没有此限制，
	 * 因此每个发布操作都有一个ExprState。
	 */
	ExprState  *exprstate[NUM_ROWFILTER_PUBACTIONS];
	EState	   *estate;			/* 用于行过滤的执行器状态 */
	TupleTableSlot *new_slot;	/* 存储新元组的槽 */
	TupleTableSlot *old_slot;	/* 存储旧元组的槽 */

	/*
	 * 发布更改的关系的OID。对于一个分区，如果为发布设置了publish_via_partition_root，
	 * 这可以设置为其祖先之一，复制更改时将使用其模式。
	 */
	Oid			publish_as_relid;

	/*
	 * 在使用祖先的模式进行复制时使用的映射，将元组从分区的类型转换为祖先的类型；
	 * 如果publish_as_relid与'relid'相同或由于分区和祖先具有相同的TupleDesc而不必要，则为NULL。
	 */
	AttrMap    *attrmap;

	/*
	 * 包含在发布中的列，或如果所有列隐式包括则为NULL。
	 * 请注意，此位图中的attnums未通过FirstLowInvalidHeapAttributeNumber进行移位。
	 */
	Bitmapset  *columns;

	/*
	 * 存储此条目附加数据的私有上下文 - 行过滤表达式的状态、列列表等。
	 */
	MemoryContext entry_cxt;
} RelationSyncEntry;

/*
 * 维护一个每个事务级别的变量以跟踪事务是否已经发送 BEGIN。
 * 仅在处理事务中的第一个更改时发送 BEGIN。这使得可以跳过发送空事务的 BEGIN/COMMIT
 * 消息对，从而节省网络带宽。
 *
 * 此优化不适用于准备好的事务，因为如果 WALSender 在事务准备之后和同一事务的提交准备之前重新启动，
 * 我们将无法判断是否因为事务为空而跳过了发送 BEGIN/PREPARE。这是因为我们将失去在重新启动之前存在的内存中的 txndata 信息。
 * 这将导致发送一个虚假的 COMMIT PREPARED，而没有对应的准备事务，这在下游处理时会导致错误。
 *
 * XXX 我们可以通过改变协议以发送附加信息来实现此优化，以便下游能够检测到相应的准备没有被发送。
 * 然而，为每个事务在下游添加这样的检查可能代价高昂，因此我们可能希望将其作为可选项进行处理。
 *
 * 对于流式事务，我们也没有这个优化，因为它们可能包含准备好的事务。
 */
typedef struct PGOutputTxnData
{
	bool		sent_begin_txn; /* 标记表示是否已发送 BEGIN */
} PGOutputTxnData;

/* 用于记住我们发送了哪些关系模式的映射。 */
static HTAB *RelationSyncCache = NULL;

static void fc_init_rel_sync_cache(MemoryContext fc_decoding_context);
static void fc_cleanup_rel_sync_cache(TransactionId fc_xid, bool fc_is_commit);
static RelationSyncEntry *fc_get_rel_sync_entry(PGOutputData *fc_data,
											 Relation fc_relation);
static void fc_rel_sync_cache_relation_cb(Datum fc_arg, Oid fc_relid);
static void fc_rel_sync_cache_publication_cb(Datum fc_arg, int fc_cacheid,
										  uint32 fc_hashvalue);
static void fc_set_schema_sent_in_streamed_txn(RelationSyncEntry *fc_entry,
											TransactionId fc_xid);
static bool fc_get_schema_sent_in_streamed_txn(RelationSyncEntry *fc_entry,
											TransactionId fc_xid);
static void fc_init_tuple_slot(PGOutputData *fc_data, Relation fc_relation,
							RelationSyncEntry *fc_entry);

/* 行过滤例程 */
static EState *fc_create_estate_for_relation(Relation fc_rel);
static void fc_pgoutput_row_filter_init(PGOutputData *fc_data,
									 List *fc_publications,
									 RelationSyncEntry *fc_entry);
static bool fc_pgoutput_row_filter_exec_expr(ExprState *fc_state,
										  ExprContext *fc_econtext);
static bool fc_pgoutput_row_filter(Relation fc_relation, TupleTableSlot *fc_old_slot,
								TupleTableSlot **fc_new_slot_ptr,
								RelationSyncEntry *fc_entry,
								ReorderBufferChangeType *fc_action);

/* 列表例程 */
static void fc_pgoutput_column_list_init(PGOutputData *fc_data,
									  List *fc_publications,
									  RelationSyncEntry *fc_entry);

/*
 * 指定输出插件回调
 */
void _PG_output_plugin_init(OutputPluginCallbacks *fc_cb)
{
	AssertVariableIsOfType(&_PG_output_plugin_init, LogicalOutputPluginInit);

	fc_cb->startup_cb = fc_pgoutput_startup;
	fc_cb->begin_cb = fc_pgoutput_begin_txn;
	fc_cb->change_cb = fc_pgoutput_change;
	fc_cb->truncate_cb = fc_pgoutput_truncate;
	fc_cb->message_cb = fc_pgoutput_message;
	fc_cb->commit_cb = fc_pgoutput_commit_txn;

	fc_cb->begin_prepare_cb = fc_pgoutput_begin_prepare_txn;
	fc_cb->prepare_cb = fc_pgoutput_prepare_txn;
	fc_cb->commit_prepared_cb = fc_pgoutput_commit_prepared_txn;
	fc_cb->rollback_prepared_cb = fc_pgoutput_rollback_prepared_txn;
	fc_cb->filter_by_origin_cb = fc_pgoutput_origin_filter;
	fc_cb->shutdown_cb = fc_pgoutput_shutdown;

	/* 事务流式传输 */
	fc_cb->stream_start_cb = fc_pgoutput_stream_start;
	fc_cb->stream_stop_cb = fc_pgoutput_stream_stop;
	fc_cb->stream_abort_cb = fc_pgoutput_stream_abort;
	fc_cb->stream_commit_cb = fc_pgoutput_stream_commit;
	fc_cb->stream_change_cb = fc_pgoutput_change;
	fc_cb->stream_message_cb = fc_pgoutput_message;
	fc_cb->stream_truncate_cb = fc_pgoutput_truncate;
	/* 事务流式传输 - 两阶段提交 */
	fc_cb->stream_prepare_cb = fc_pgoutput_stream_prepare_txn;
}

static void fc_parse_output_parameters(List *fc_options, PGOutputData *fc_data)
{
	ListCell   *fc_lc;
	bool		fc_protocol_version_given = false;
	bool		fc_publication_names_given = false;
	bool		fc_binary_option_given = false;
	bool		fc_messages_option_given = false;
	bool		fc_streaming_given = false;
	bool		fc_two_phase_option_given = false;

	fc_data->binary = false;
	fc_data->streaming = false;
	fc_data->messages = false;
	fc_data->two_phase = false;

	foreach(fc_lc, fc_options)
	{
		DefElem    *fc_defel = (DefElem *) lfirst(fc_lc);

		Assert(fc_defel->arg == NULL || IsA(fc_defel->arg, String));

		/* 检查每个参数，我们是否识别它 */
		if (strcmp(fc_defel->defname, "proto_version") == 0)
		{
			unsigned long fc_parsed;
			char	   *fc_endptr;

			if (fc_protocol_version_given)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("conflicting or redundant options")));
			fc_protocol_version_given = true;

			errno = 0;
			fc_parsed = strtoul(strVal(fc_defel->arg), &fc_endptr, 10);
			if (errno != 0 || *fc_endptr != '\0')
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("invalid proto_version")));

			if (fc_parsed > PG_UINT32_MAX)
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("proto_version \"%s\" out of range",
								strVal(fc_defel->arg))));

			fc_data->protocol_version = (uint32) fc_parsed;
		}
		else if (strcmp(fc_defel->defname, "publication_names") == 0)
		{
			if (fc_publication_names_given)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("conflicting or redundant options")));
			fc_publication_names_given = true;

			if (!SplitIdentifierString(strVal(fc_defel->arg), ',',
									   &fc_data->publication_names))
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_NAME),
						 errmsg("invalid publication_names syntax")));
		}
		else if (strcmp(fc_defel->defname, "binary") == 0)
		{
			if (fc_binary_option_given)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("conflicting or redundant options")));
			fc_binary_option_given = true;

			fc_data->binary = defGetBoolean(fc_defel);
		}
		else if (strcmp(fc_defel->defname, "messages") == 0)
		{
			if (fc_messages_option_given)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("conflicting or redundant options")));
			fc_messages_option_given = true;

			fc_data->messages = defGetBoolean(fc_defel);
		}
		else if (strcmp(fc_defel->defname, "streaming") == 0)
		{
			if (fc_streaming_given)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("conflicting or redundant options")));
			fc_streaming_given = true;

			fc_data->streaming = defGetBoolean(fc_defel);
		}
		else if (strcmp(fc_defel->defname, "two_phase") == 0)
		{
			if (fc_two_phase_option_given)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("conflicting or redundant options")));
			fc_two_phase_option_given = true;

			fc_data->two_phase = defGetBoolean(fc_defel);
		}
		else
			elog(ERROR, "unrecognized pgoutput option: %s", fc_defel->defname);
	}
}

/*
 * PGOutputData->context 的回调，负责清理 pubctx。
 */
static void fc_pgoutput_pubctx_reset_callback(void *fc_arg)
{
	pubctx = NULL;
}

/*
 * 初始化此插件
 */
static void fc_pgoutput_startup(LogicalDecodingContext *fc_ctx, OutputPluginOptions *fc_opt,
				 bool fc_is_init)
{
	PGOutputData *fc_data = palloc0(sizeof(PGOutputData));
	static bool fc_publication_callback_registered = false;
	MemoryContextCallback *fc_mcallback;

	/* 为私有分配创建我们的内存上下文。 */
	fc_data->context = AllocSetContextCreate(fc_ctx->context,
										  "logical replication output context",
										  ALLOCSET_DEFAULT_SIZES);

	fc_data->cachectx = AllocSetContextCreate(fc_ctx->context,
										   "logical replication cache context",
										   ALLOCSET_DEFAULT_SIZES);

	Assert(pubctx == NULL);
	pubctx = AllocSetContextCreate(fc_ctx->context,
								   "logical replication publication list context",
								   ALLOCSET_SMALL_SIZES);

	fc_mcallback = palloc0(sizeof(MemoryContextCallback));
	fc_mcallback->func = fc_pgoutput_pubctx_reset_callback;
	MemoryContextRegisterResetCallback(fc_ctx->context, fc_mcallback);

	fc_ctx->output_plugin_private = fc_data;

	/* 此插件使用二进制协议。 */
	fc_opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;

	/*
	 * 这是复制开始，而不是插槽初始化。
	 *
	 * 解析和验证客户端传递的选项。
	 */
	if (!fc_is_init)
	{
		/* 解析参数，如果发现任何不识别的则错误 */
		fc_parse_output_parameters(fc_ctx->output_plugin_options, fc_data);

		/* 检查我们是否支持请求的协议 */
		if (fc_data->protocol_version > LOGICALREP_PROTO_MAX_VERSION_NUM)
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("client sent proto_version=%d but we only support protocol %d or lower",
							fc_data->protocol_version, LOGICALREP_PROTO_MAX_VERSION_NUM)));

		if (fc_data->protocol_version < LOGICALREP_PROTO_MIN_VERSION_NUM)
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("client sent proto_version=%d but we only support protocol %d or higher",
							fc_data->protocol_version, LOGICALREP_PROTO_MIN_VERSION_NUM)));

		if (list_length(fc_data->publication_names) < 1)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("publication_names parameter missing")));

		/*
		 * 决定是否启用流式处理。默认情况下禁用，在这种情况下我们只是更新解码上下文中的标志。
		 * 否则，我们仅在协议具有足够版本时允许，并且输出插件支持它时允许。
		 */
		if (!fc_data->streaming)
			fc_ctx->streaming = false;
		else if (fc_data->protocol_version < LOGICALREP_PROTO_STREAM_VERSION_NUM)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("requested proto_version=%d does not support streaming, need %d or higher",
							fc_data->protocol_version, LOGICALREP_PROTO_STREAM_VERSION_NUM)));
		else if (!fc_ctx->streaming)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("streaming requested, but not supported by output plugin")));

		/* 还要记住我们当前未流式传输任何事务。 */
		in_streaming = false;

		/*
		 * 在这里，我们只是检查插件是否传递了两阶段选项，并决定是否在稍后启用它。
		 * 如果之前的启动已经启用，则保持启用。但我们仅在协议具有足够版本时允许传递选项，并且当输出插件支持它时允许。
		 */
		if (!fc_data->two_phase)
			fc_ctx->twophase_opt_given = false;
		else if (fc_data->protocol_version < LOGICALREP_PROTO_TWOPHASE_VERSION_NUM)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("requested proto_version=%d does not support two-phase commit, need %d or higher",
							fc_data->protocol_version, LOGICALREP_PROTO_TWOPHASE_VERSION_NUM)));
		else if (!fc_ctx->twophase)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("two-phase commit requested, but not supported by output plugin")));
		else
			fc_ctx->twophase_opt_given = true;

		/* 初始化发布状态。 */
		fc_data->publications = NIL;
		publications_valid = false;

		/*
		 * 如果在该进程的某个先前调用中尚未注册pg_publication的回调，则进行注册。
		 */
		if (!fc_publication_callback_registered)
		{
			CacheRegisterSyscacheCallback(PUBLICATIONOID,
										  fc_publication_invalidation_cb,
										  (Datum) 0);
			fc_publication_callback_registered = true;
		}

		/* 初始化关系模式缓存。 */
		fc_init_rel_sync_cache(CacheMemoryContext);
	}
	else
	{
		/*
		 * 在槽初始化模式下禁用流式传输和预提交事务。
		 */
		fc_ctx->streaming = false;
		fc_ctx->twophase = false;
	}
}

/*
 * BEGIN回调。
 *
 * 不要在这里发送BEGIN消息，而是推迟到第一次更改。 在逻辑复制中，常见场景是复制一组表（而不是所有表），并且在未发布的表上进行更改的事务会产生空事务。 这些空事务将向订阅者发送BEGIN和COMMIT消息，使用带宽在逻辑复制中几乎没有用处的东西。
 */
static void fc_pgoutput_begin_txn(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn)
{
	PGOutputTxnData *fc_txndata = MemoryContextAllocZero(fc_ctx->context,
													  sizeof(PGOutputTxnData));

	fc_txn->output_plugin_private = fc_txndata;
}

/*
 * 发送BEGIN。
 *
 * 这是在处理事务的第一次更改时调用的。
 */
static void fc_pgoutput_send_begin(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn)
{
	bool		fc_send_replication_origin = fc_txn->origin_id != InvalidRepOriginId;
	PGOutputTxnData *fc_txndata = (PGOutputTxnData *) fc_txn->output_plugin_private;

	Assert(fc_txndata);
	Assert(!fc_txndata->sent_begin_txn);

	OutputPluginPrepareWrite(fc_ctx, !fc_send_replication_origin);
	logicalrep_write_begin(fc_ctx->out, fc_txn);
	fc_txndata->sent_begin_txn = true;

	fc_send_repl_origin(fc_ctx, fc_txn->origin_id, fc_txn->origin_lsn,
					 fc_send_replication_origin);

	OutputPluginWrite(fc_ctx, true);
}

/*
 * COMMIT回调
 */
static void fc_pgoutput_commit_txn(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
					XLogRecPtr fc_commit_lsn)
{
	PGOutputTxnData *fc_txndata = (PGOutputTxnData *) fc_txn->output_plugin_private;
	bool		fc_sent_begin_txn;

	Assert(fc_txndata);

	/*
	 * 除非已经向下游发送了该事务的某些相关更改，否则我们不需要发送提交消息。
	 */
	fc_sent_begin_txn = fc_txndata->sent_begin_txn;
	fc_update_replication_progress(fc_ctx, !fc_sent_begin_txn);
	pfree(fc_txndata);
	fc_txn->output_plugin_private = NULL;

	if (!fc_sent_begin_txn)
	{
		elog(DEBUG1, "skipped replication of an empty transaction with XID: %u", fc_txn->xid);
		return;
	}

	OutputPluginPrepareWrite(fc_ctx, true);
	logicalrep_write_commit(fc_ctx->out, fc_txn, fc_commit_lsn);
	OutputPluginWrite(fc_ctx, true);
}

/*
 * BEGIN PREPARE回调
 */
static void fc_pgoutput_begin_prepare_txn(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn)
{
	bool		fc_send_replication_origin = fc_txn->origin_id != InvalidRepOriginId;

	OutputPluginPrepareWrite(fc_ctx, !fc_send_replication_origin);
	logicalrep_write_begin_prepare(fc_ctx->out, fc_txn);

	fc_send_repl_origin(fc_ctx, fc_txn->origin_id, fc_txn->origin_lsn,
					 fc_send_replication_origin);

	OutputPluginWrite(fc_ctx, true);
}

/*
 * PREPARE回调
 */
static void fc_pgoutput_prepare_txn(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
					 XLogRecPtr fc_prepare_lsn)
{
	fc_update_replication_progress(fc_ctx, false);

	OutputPluginPrepareWrite(fc_ctx, true);
	logicalrep_write_prepare(fc_ctx->out, fc_txn, fc_prepare_lsn);
	OutputPluginWrite(fc_ctx, true);
}

/*
 * COMMIT PREPARED回调
 */
static void fc_pgoutput_commit_prepared_txn(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
							 XLogRecPtr fc_commit_lsn)
{
	fc_update_replication_progress(fc_ctx, false);

	OutputPluginPrepareWrite(fc_ctx, true);
	logicalrep_write_commit_prepared(fc_ctx->out, fc_txn, fc_commit_lsn);
	OutputPluginWrite(fc_ctx, true);
}

/*
 * ROLLBACK PREPARED回调
 */
static void fc_pgoutput_rollback_prepared_txn(LogicalDecodingContext *fc_ctx,
							   ReorderBufferTXN *fc_txn,
							   XLogRecPtr fc_prepare_end_lsn,
							   TimestampTz fc_prepare_time)
{
	fc_update_replication_progress(fc_ctx, false);

	OutputPluginPrepareWrite(fc_ctx, true);
	logicalrep_write_rollback_prepared(fc_ctx->out, fc_txn, fc_prepare_end_lsn,
									   fc_prepare_time);
	OutputPluginWrite(fc_ctx, true);
}

/*
 * 如果尚未完成，则写入关系及其祖先（如果有）的当前模式。
 */
static void fc_maybe_send_schema(LogicalDecodingContext *fc_ctx,
				  ReorderBufferChange *fc_change,
				  Relation fc_relation, RelationSyncEntry *fc_relentry)
{
	bool		fc_schema_sent;
	TransactionId fc_xid = InvalidTransactionId;
	TransactionId fc_topxid = InvalidTransactionId;

	/*
	 * 记住变更的（子）事务的XID。 我们不关心它是否是顶层事务（我们已经在当前流块的开始发送了该XID）。
	 *
	 * 如果我们不在流块中，只需使用InvalidTransactionId，写入方法将不会包含它。
	 */
	if (in_streaming)
		fc_xid = fc_change->txn->xid;

	if (fc_change->txn->toptxn)
		fc_topxid = fc_change->txn->toptxn->xid;
	else
		fc_topxid = fc_xid;

	/*
	 * 我们需要发送模式吗？ 我们确实单独跟踪流式事务，因为这些可能在稍后应用（常规事务在那之前不会看到它们的影响），并且顺序我们目前不知道。
	 *
	 * XXX 这里有优化的空间。 目前，我们在流式事务中总是第一次发送模式，但我们可能通过检查'relentry->schema_sent'标志来避免这一点。 不过，在这样做之前，我们需要研究其对混合流式和非流式事务的影响。
	 */
	if (in_streaming)
		fc_schema_sent = fc_get_schema_sent_in_streamed_txn(fc_relentry, fc_topxid);
	else
		fc_schema_sent = fc_relentry->schema_sent;

	/* 如果我们已经发送了模式，就不需要做任何事情。 */
	if (fc_schema_sent)
		return;

	/*
	 * 发送模式。 如果更改将使用祖先的模式而不是关系自己的模式发布，请先发送该祖先的模式，然后再发送关系自己的模式（XXX - 也许仅发送前者就足够？）。
	 */
	if (fc_relentry->publish_as_relid != RelationGetRelid(fc_relation))
	{
		Relation	fc_ancestor = RelationIdGetRelation(fc_relentry->publish_as_relid);

		fc_send_relation_and_attrs(fc_ancestor, fc_xid, fc_ctx, fc_relentry->columns);
		RelationClose(fc_ancestor);
	}

	fc_send_relation_and_attrs(fc_relation, fc_xid, fc_ctx, fc_relentry->columns);

	if (in_streaming)
		fc_set_schema_sent_in_streamed_txn(fc_relentry, fc_topxid);
	else
		fc_relentry->schema_sent = true;
}

/*
 * 发送关系
 */
static void fc_send_relation_and_attrs(Relation fc_relation, TransactionId fc_xid,
						LogicalDecodingContext *fc_ctx,
						Bitmapset *fc_columns)
{
	TupleDesc	fc_desc = RelationGetDescr(fc_relation);
	int			fc_i;

	/*
	 * 如果需要，写出类型信息。 我们仅对用户创建的类型执行此操作。 我们使用FirstGenbkiObjectId作为界限，因此我们只考虑手动分配OID的对象被视为“内置”，而不是例如在information_schema中定义的任何函数或类型。 这很重要，因为只有手动分配的OID才能被期望在主要版本之间保持稳定。
	 */
	for (fc_i = 0; fc_i < fc_desc->natts; fc_i++)
	{
		Form_pg_attribute fc_att = TupleDescAttr(fc_desc, fc_i);

		if (fc_att->attisdropped || fc_att->attgenerated)
			continue;

		if (fc_att->atttypid < FirstGenbkiObjectId)
			continue;

		/* 如果在列列表中不存在，则跳过此属性 */
		if (fc_columns != NULL && !bms_is_member(fc_att->attnum, fc_columns))
			continue;

		OutputPluginPrepareWrite(fc_ctx, false);
		logicalrep_write_typ(fc_ctx->out, fc_xid, fc_att->atttypid);
		OutputPluginWrite(fc_ctx, false);
	}

	OutputPluginPrepareWrite(fc_ctx, false);
	logicalrep_write_rel(fc_ctx->out, fc_xid, fc_relation, fc_columns);
	OutputPluginWrite(fc_ctx, false);
}

/*
 * 执行器状态为评估指定关系的行过滤器表达式的准备。
 */
static EState *
fc_create_estate_for_relation(Relation fc_rel)
{
	EState	   *fc_estate;
	RangeTblEntry *fc_rte;

	fc_estate = CreateExecutorState();

	fc_rte = makeNode(RangeTblEntry);
	fc_rte->rtekind = RTE_RELATION;
	fc_rte->relid = RelationGetRelid(fc_rel);
	fc_rte->relkind = fc_rel->rd_rel->relkind;
	fc_rte->rellockmode = AccessShareLock;
	ExecInitRangeTable(fc_estate, list_make1(fc_rte));

	fc_estate->es_output_cid = GetCurrentCommandId(false);

	return fc_estate;
}

/*
 * 评估行过滤器。
 *
 * 如果行过滤器评估为 NULL，则视为假，即更改未被复制。
 */
static bool fc_pgoutput_row_filter_exec_expr(ExprState *fc_state, ExprContext *fc_econtext)
{
	Datum		fc_ret;
	bool		fc_isnull;

	Assert(fc_state != NULL);

	fc_ret = ExecEvalExprSwitchContext(fc_state, fc_econtext, &fc_isnull);

	elog(DEBUG3, "row filter evaluates to %s (isnull: %s)",
		 fc_isnull ? "false" : DatumGetBool(fc_ret) ? "true" : "false",
		 fc_isnull ? "true" : "false");

	if (fc_isnull)
		return false;

	return DatumGetBool(fc_ret);
}

/*
 * 确保每个条目的内存上下文存在。
 */
static void fc_pgoutput_ensure_entry_cxt(PGOutputData *fc_data, RelationSyncEntry *fc_entry)
{
	Relation	fc_relation;

	/* 上下文可能已经存在，在这种情况下退出。 */
	if (fc_entry->entry_cxt)
		return;

	fc_relation = RelationIdGetRelation(fc_entry->publish_as_relid);

	fc_entry->entry_cxt = AllocSetContextCreate(fc_data->cachectx,
											 "entry private context",
											 ALLOCSET_SMALL_SIZES);

	MemoryContextCopyAndSetIdentifier(fc_entry->entry_cxt,
									  RelationGetRelationName(fc_relation));
}

/*
 * 初始化行过滤器。
 */
static void fc_pgoutput_row_filter_init(PGOutputData *fc_data, List *fc_publications,
						 RelationSyncEntry *fc_entry)
{
	ListCell   *fc_lc;
	List	   *fc_rfnodes[] = {NIL, NIL, NIL};	/* 每个 pubaction 一个 */
	bool		fc_no_filter[] = {false, false, false};	/* 每个 pubaction 一个 */
	MemoryContext fc_oldctx;
	int			fc_idx;
	bool		fc_has_filter = true;
	Oid			fc_schemaid = get_rel_namespace(fc_entry->publish_as_relid);

	/*
	 * 找出这个关系是否有任何行过滤器。如果有，则准备必要的 ExprState 并将其缓存到 entry->exprstate 中。要构建表达式状态，我们需要确保以下几点：
	 *
	 * 必须检查所有给定的发布-表映射。
	 *
	 * 多个发布可能对这个关系有多个行过滤器。由于行过滤器的使用取决于 DML 操作，因此有多个列表（每个操作一个）将附加行过滤器。
	 *
	 * FOR ALL TABLES 和 FOR TABLES IN SCHEMA 意味着“不要使用行过滤器表达式”，因此其优先级更高。
	 */
	foreach(fc_lc, fc_publications)
	{
		Publication *fc_pub = lfirst(fc_lc);
		HeapTuple	fc_rftuple = NULL;
		Datum		fc_rfdatum = 0;
		bool		fc_pub_no_filter = true;

		/*
		 * 如果发布是 FOR ALL TABLES，或者发布包含一个 FOR TABLES IN SCHEMA 其中表属于被引用的模式，则将其视为没有行过滤器的情况（即使其他发布有行过滤器）。
		 */
		if (!fc_pub->alltables &&
			!SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
								   ObjectIdGetDatum(fc_schemaid),
								   ObjectIdGetDatum(fc_pub->oid)))
		{
			/*
			 * 检查此发布中是否存在行过滤器。
			 */
			fc_rftuple = SearchSysCache2(PUBLICATIONRELMAP,
									  ObjectIdGetDatum(fc_entry->publish_as_relid),
									  ObjectIdGetDatum(fc_pub->oid));

			if (HeapTupleIsValid(fc_rftuple))
			{
				/* NULL 表示没有过滤器。 */
				fc_rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, fc_rftuple,
										  Anum_pg_publication_rel_prqual,
										  &fc_pub_no_filter);
			}
		}

		if (fc_pub_no_filter)
		{
			if (fc_rftuple)
				ReleaseSysCache(fc_rftuple);

			fc_no_filter[PUBACTION_INSERT] |= fc_pub->pubactions.pubinsert;
			fc_no_filter[PUBACTION_UPDATE] |= fc_pub->pubactions.pubupdate;
			fc_no_filter[PUBACTION_DELETE] |= fc_pub->pubactions.pubdelete;

			/*
			 * 如果所有 DML 操作通过该发布公开，则快速退出。
			 */
			if (fc_no_filter[PUBACTION_INSERT] &&
				fc_no_filter[PUBACTION_UPDATE] &&
				fc_no_filter[PUBACTION_DELETE])
			{
				fc_has_filter = false;
				break;
			}

			/* 此发布没有额外的工作。下一个。 */
			continue;
		}

		/* 形成每个 pubaction 的行过滤器列表。 */
		if (fc_pub->pubactions.pubinsert && !fc_no_filter[PUBACTION_INSERT])
			fc_rfnodes[PUBACTION_INSERT] = lappend(fc_rfnodes[PUBACTION_INSERT],
												TextDatumGetCString(fc_rfdatum));
		if (fc_pub->pubactions.pubupdate && !fc_no_filter[PUBACTION_UPDATE])
			fc_rfnodes[PUBACTION_UPDATE] = lappend(fc_rfnodes[PUBACTION_UPDATE],
												TextDatumGetCString(fc_rfdatum));
		if (fc_pub->pubactions.pubdelete && !fc_no_filter[PUBACTION_DELETE])
			fc_rfnodes[PUBACTION_DELETE] = lappend(fc_rfnodes[PUBACTION_DELETE],
												TextDatumGetCString(fc_rfdatum));

		ReleaseSysCache(fc_rftuple);
	}							/* 循环遍历所有已订阅的发布 */

	/* 清理行过滤器 */
	for (fc_idx = 0; fc_idx < NUM_ROWFILTER_PUBACTIONS; fc_idx++)
	{
		if (fc_no_filter[fc_idx])
		{
			list_free_deep(fc_rfnodes[fc_idx]);
			fc_rfnodes[fc_idx] = NIL;
		}
	}

	if (fc_has_filter)
	{
		Relation	fc_relation = RelationIdGetRelation(fc_entry->publish_as_relid);

		fc_pgoutput_ensure_entry_cxt(fc_data, fc_entry);

		/*
		 * 现在所有 pubaction 的过滤器都已知晓。当它们的 pubaction 相同时，合并它们。
		 */
		fc_oldctx = MemoryContextSwitchTo(fc_entry->entry_cxt);
		fc_entry->estate = fc_create_estate_for_relation(fc_relation);
		for (fc_idx = 0; fc_idx < NUM_ROWFILTER_PUBACTIONS; fc_idx++)
		{
			List	   *fc_filters = NIL;
			Expr	   *fc_rfnode;

			if (fc_rfnodes[fc_idx] == NIL)
				continue;

			foreach(fc_lc, fc_rfnodes[fc_idx])
				fc_filters = lappend(fc_filters, stringToNode((char *) lfirst(fc_lc)));

			/* 合并行过滤器并缓存 ExprState */
			fc_rfnode = make_orclause(fc_filters);
			fc_entry->exprstate[fc_idx] = ExecPrepareExpr(fc_rfnode, fc_entry->estate);
		}						/* 对于每个 pubaction */
		MemoryContextSwitchTo(fc_oldctx);

		RelationClose(fc_relation);
	}
}

/*
 * 初始化列列表。
 */
static void fc_pgoutput_column_list_init(PGOutputData *fc_data, List *fc_publications,
						  RelationSyncEntry *fc_entry)
{
	ListCell   *fc_lc;
	bool		fc_first = true;
	Relation	fc_relation = RelationIdGetRelation(fc_entry->publish_as_relid);

	/*
	 * 查找这个关系是否有任何列列表。如果有，则使用列列表构建位图。
	 *
	 * 多个发布可能对这个关系有多个列列表。
	 *
	 * 注意，我们不支持在组合发布时同一表的列列表不同的情况。请查看 fetch_table_list 顶部的注释。但可以稍后更改发布，因此我们仍然需要检查所有给定的发布-表映射，并在任何发布有不同列列表时报告错误。
	 *
	 * FOR ALL TABLES 和 FOR TABLES IN SCHEMA 暗示“不要使用列列表”。
	 */
	foreach(fc_lc, fc_publications)
	{
		Publication *fc_pub = lfirst(fc_lc);
		HeapTuple	fc_cftuple = NULL;
		Datum		fc_cfdatum = 0;
		Bitmapset  *fc_cols = NULL;

		/*
		 * 如果发布是 FOR ALL TABLES，则视为没有列列表的情况（即使其他发布有列表）。
		 */
		if (!fc_pub->alltables)
		{
			bool		fc_pub_no_list = true;

			/*
			 * 检查该发布中是否存在列列表。
			 *
			 * 注意：如果我们没有找到 pg_publication_rel 行，这意味着这是一个
			 * 为整个模式定义的发布，因此不能有列列表，
			 * 就像 FOR ALL TABLES 发布一样。
			 */
			fc_cftuple = SearchSysCache2(PUBLICATIONRELMAP,
									  ObjectIdGetDatum(fc_entry->publish_as_relid),
									  ObjectIdGetDatum(fc_pub->oid));

			if (HeapTupleIsValid(fc_cftuple))
			{
				/* 查找列列表属性。 */
				fc_cfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, fc_cftuple,
										  Anum_pg_publication_rel_prattrs,
										  &fc_pub_no_list);

				/* 在每个条目的上下文中构建列列表位图。 */
				if (!fc_pub_no_list)	/* 当不为 null 时 */
				{
					fc_pgoutput_ensure_entry_cxt(fc_data, fc_entry);

					fc_cols = pub_collist_to_bitmapset(fc_cols, fc_cfdatum,
													fc_entry->entry_cxt);

					/*
					 * 如果列列表包括表的所有列，
					 * 将其设置为 NULL。
					 */
					if (bms_num_members(fc_cols) == RelationGetNumberOfAttributes(fc_relation))
					{
						bms_free(fc_cols);
						fc_cols = NULL;
					}
				}

				ReleaseSysCache(fc_cftuple);
			}
		}

		if (fc_first)
		{
			fc_entry->columns = fc_cols;
			fc_first = false;
		}
		else if (!bms_equal(fc_entry->columns, fc_cols))
			ereport(ERROR,
					errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					errmsg("cannot use different column lists for table \"%s.%s\" in different publications",
						   get_namespace_name(RelationGetNamespace(fc_relation)),
						   RelationGetRelationName(fc_relation)));
	}							/* 循环遍历所有已订阅的发布 */

	RelationClose(fc_relation);
}

/*
 * 初始化用于存储新旧元组的槽，并构建一个映射，
 * 用于将关系的元组转换为祖先的格式。
 */
static void fc_init_tuple_slot(PGOutputData *fc_data, Relation fc_relation,
				RelationSyncEntry *fc_entry)
{
	MemoryContext fc_oldctx;
	TupleDesc	fc_oldtupdesc;
	TupleDesc	fc_newtupdesc;

	fc_oldctx = MemoryContextSwitchTo(fc_data->cachectx);

	/*
	 * 创建元组表槽。创建 TupleDesc 的副本，因为它需要
	 * 保持与缓存一样长的时间。
	 */
	fc_oldtupdesc = CreateTupleDescCopyConstr(RelationGetDescr(fc_relation));
	fc_newtupdesc = CreateTupleDescCopyConstr(RelationGetDescr(fc_relation));

	fc_entry->old_slot = MakeSingleTupleTableSlot(fc_oldtupdesc, &TTSOpsHeapTuple);
	fc_entry->new_slot = MakeSingleTupleTableSlot(fc_newtupdesc, &TTSOpsHeapTuple);

	MemoryContextSwitchTo(fc_oldctx);

	/*
	 * 缓存将用于将关系的元组转换为
	 * 祖先格式的映射（如果需要）。
	 */
	if (fc_entry->publish_as_relid != RelationGetRelid(fc_relation))
	{
		Relation	fc_ancestor = RelationIdGetRelation(fc_entry->publish_as_relid);
		TupleDesc	fc_indesc = RelationGetDescr(fc_relation);
		TupleDesc	fc_outdesc = RelationGetDescr(fc_ancestor);

		/* 映射必须与逻辑解码上下文的生命周期相同。 */
		fc_oldctx = MemoryContextSwitchTo(fc_data->cachectx);

		fc_entry->attrmap = build_attrmap_by_name_if_req(fc_indesc, fc_outdesc);

		MemoryContextSwitchTo(fc_oldctx);
		RelationClose(fc_ancestor);
	}
}

/*
 * 更改如果有行过滤器会被检查。
 *
 * 如果更改需要被复制，则返回true，否则返回false。
 *
 * 对于插入，评估新元组的行过滤器。
 * 对于删除，评估旧元组的行过滤器。
 * 对于更新，评估旧元组和新元组的行过滤器。
 *
 * 对于更新，如果两个评估都为真，我们允许发送UPDATE，
 * 如果两个评估都为假，则不复制UPDATE。现在，如果只有
 * 一个元组匹配行过滤器表达式，我们会根据以下规则将
 * UPDATE转换为DELETE或INSERT，以避免数据不一致：
 *
 * 情况 1：旧行（不匹配）    新行（不匹配）  -> （丢弃更改）
 * 情况 2：旧行（不匹配）    新行（匹配）     -> INSERT
 * 情况 3：旧行（匹配）       新行（不匹配）  -> DELETE
 * 情况 4：旧行（匹配）       新行（匹配）     -> UPDATE
 *
 * 新的操作会在操作参数中更新。
 *
 * 当将UPDATE转换为INSERT时，新的槽位可能会被更新，
 * 因为原始的新元组可能没有来自副本标识的列值。
 *
 * 示例：
 * 假设旧元组满足行过滤器，但新元组不满足。
 * 由于旧元组满足，初始表同步复制了这一行
 * （或者使用其他方法确保数据的一致性）。然而，在UPDATE之后，新元组不满足行过滤器，
 * 因此从数据一致性的角度来看，该行应该在订阅者上被删除。
 * UPDATE应该转换为DELETE语句并发送到订阅者。
 * 在订阅者上保留这一行是不理想的，因为它不反映
 * 发布者行过滤器表达式中定义的内容。这个行在订阅者上很可能不会
 * 再次被复制修改。如果有人插入了一行具有相同旧标识符的新行，
 * 复制可能会因约束冲突而停止。
 *
 * 假设旧元组不匹配行过滤器，但新元组匹配。
 * 由于旧元组不满足，初始表同步
 * 可能没有复制这一行。然而，在UPDATE之后，新元组确实
 * 满足行过滤器，因此从数据一致性的角度来看，该行
 * 应该在订阅者上被插入。否则，后续的UPDATE或DELETE
 * 语句将没有效果（它不匹配任何行 -- 参见
 * apply_handle_update_internal()）。所以，UPDATE应该转换为
 * INSERT语句并发送到订阅者。然而，这可能会让期待数据集
 * 满足提供者行过滤器表达式的人感到惊讶。
 */
static bool fc_pgoutput_row_filter(Relation fc_relation, TupleTableSlot *fc_old_slot,
					TupleTableSlot **fc_new_slot_ptr, RelationSyncEntry *fc_entry,
					ReorderBufferChangeType *fc_action)
{
	TupleDesc	fc_desc;
	int			fc_i;
	bool		fc_old_matched,
				fc_new_matched,
				fc_result;
	TupleTableSlot *fc_tmp_new_slot;
	TupleTableSlot *fc_new_slot = *fc_new_slot_ptr;
	ExprContext *fc_ecxt;
	ExprState  *fc_filter_exprstate;

	/*
	 * 我们需要这个映射以避免依赖ReorderBufferChangeType枚举
	 * 拥有特定值。
	 */
	static const int fc_map_changetype_pubaction[] = {
		[REORDER_BUFFER_CHANGE_INSERT] = PUBACTION_INSERT,
		[REORDER_BUFFER_CHANGE_UPDATE] = PUBACTION_UPDATE,
		[REORDER_BUFFER_CHANGE_DELETE] = PUBACTION_DELETE
	};

	Assert(*fc_action == REORDER_BUFFER_CHANGE_INSERT ||
		   *fc_action == REORDER_BUFFER_CHANGE_UPDATE ||
		   *fc_action == REORDER_BUFFER_CHANGE_DELETE);

	Assert(fc_new_slot || fc_old_slot);

	/* 获取相应的行过滤器 */
	fc_filter_exprstate = fc_entry->exprstate[fc_map_changetype_pubaction[*fc_action]];

	/* 如果没有行过滤器则退出 */
	if (!fc_filter_exprstate)
		return true;

	elog(DEBUG3, "table \"%s.%s\" has row filter",
		 get_namespace_name(RelationGetNamespace(fc_relation)),
		 RelationGetRelationName(fc_relation));

	ResetPerTupleExprContext(fc_entry->estate);

	fc_ecxt = GetPerTupleExprContext(fc_entry->estate);

	/*
	 * 对于只有一个元组的以下情况，我们可以
	 * 对该元组评估行过滤器并返回。
	 *
	 * 对于插入，我们只有新元组。
	 *
	 * 对于更新，当副本身份列没有变化且这些列没有外部数据时，我们可能只有一个新元组，
	 * 但我们仍然需要对新元组评估行过滤器，因为这些列的现有值可能与过滤器不匹配。此外，
	 * 用户可以在行过滤器中使用常量表达式，所以我们仍然需要
	 * 对新元组进行评估。
	 *
	 * 对于删除，我们只有旧元组。
	 */
	if (!fc_new_slot || !fc_old_slot)
	{
		fc_ecxt->ecxt_scantuple = fc_new_slot ? fc_new_slot : fc_old_slot;
		fc_result = fc_pgoutput_row_filter_exec_expr(fc_filter_exprstate, fc_ecxt);

		return fc_result;
	}

	/*
	 * 只有在更新时，旧元组和新元组都必须有效，并且需要
	 * 与行过滤器进行检查。
	 */
	Assert(fc_map_changetype_pubaction[*fc_action] == PUBACTION_UPDATE);

	slot_getallattrs(fc_new_slot);
	slot_getallattrs(fc_old_slot);

	fc_tmp_new_slot = NULL;
	fc_desc = RelationGetDescr(fc_relation);

	/*
	 * 新元组可能没有所有副本身份列，在这种情况下，它需要
	 * 从旧元组复制过来。
	 */
	for (fc_i = 0; fc_i < fc_desc->natts; fc_i++)
	{
		Form_pg_attribute fc_att = TupleDescAttr(fc_desc, fc_i);

		/*
		 * 如果新元组或旧元组中的列为null，则无需处理
		 */
		if (fc_new_slot->tts_isnull[fc_i] || fc_old_slot->tts_isnull[fc_i])
			continue;

		/*
		 * 未更改的被调配副本身份列只记录在
		 * 旧元组中。将其复制到新元组中。更改（或WAL
		 * 记录）的调配值始终在内存中组装并设置为
		 * VARTAG_INDIRECT。请参见ReorderBufferToastReplace。
		 */
		if (fc_att->attlen == -1 &&
			VARATT_IS_EXTERNAL_ONDISK(fc_new_slot->tts_values[fc_i]) &&
			!VARATT_IS_EXTERNAL_ONDISK(fc_old_slot->tts_values[fc_i]))
		{
			if (!fc_tmp_new_slot)
			{
				fc_tmp_new_slot = MakeSingleTupleTableSlot(fc_desc, &TTSOpsVirtual);
				ExecClearTuple(fc_tmp_new_slot);

				memcpy(fc_tmp_new_slot->tts_values, fc_new_slot->tts_values,
					   fc_desc->natts * sizeof(Datum));
				memcpy(fc_tmp_new_slot->tts_isnull, fc_new_slot->tts_isnull,
					   fc_desc->natts * sizeof(bool));
			}

			fc_tmp_new_slot->tts_values[fc_i] = fc_old_slot->tts_values[fc_i];
			fc_tmp_new_slot->tts_isnull[fc_i] = fc_old_slot->tts_isnull[fc_i];
		}
	}

	fc_ecxt->ecxt_scantuple = fc_old_slot;
	fc_old_matched = fc_pgoutput_row_filter_exec_expr(fc_filter_exprstate, fc_ecxt);

	if (fc_tmp_new_slot)
	{
		ExecStoreVirtualTuple(fc_tmp_new_slot);
		fc_ecxt->ecxt_scantuple = fc_tmp_new_slot;
	}
	else
		fc_ecxt->ecxt_scantuple = fc_new_slot;

	fc_new_matched = fc_pgoutput_row_filter_exec_expr(fc_filter_exprstate, fc_ecxt);

	/*
	 * 情况1：如果两个元组都不匹配行过滤器，则退出。发送
	 * 什么都不发。
	 */
	if (!fc_old_matched && !fc_new_matched)
		return false;

	/*
	 * 情况2：如果旧元组不满足行过滤器，但新元组满足，则将
	 * 更新转换为插入。
	 *
	 * 使用必须包含所有副本身份列的列值的新转换元组。这是必要的，以确保在向下游节点插入元组时，
	 * 我们拥有所有所需的列值。
	 */
	if (!fc_old_matched && fc_new_matched)
	{
		*fc_action = REORDER_BUFFER_CHANGE_INSERT;

		if (fc_tmp_new_slot)
			*fc_new_slot_ptr = fc_tmp_new_slot;
	}

	/*
	 * 情况3：如果旧元组满足行过滤器但新元组不满足，则将
	 * 更新转换为删除。
	 *
	 * 此转换不需要另一个元组。旧元组将用于删除。
	 */
	else if (fc_old_matched && !fc_new_matched)
		*fc_action = REORDER_BUFFER_CHANGE_DELETE;

	/*
	 * 情况4：如果两个元组都匹配行过滤器，则不需要进行转换。
	 * (*操作为默认更新)。
	 */

	return true;
}

/*
 * 在网络上传送解码后的DML。
 *
 * 在流式和非流式模式下都调用此函数。
 */
static void fc_pgoutput_change(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
				Relation fc_relation, ReorderBufferChange *fc_change)
{
	PGOutputData *fc_data = (PGOutputData *) fc_ctx->output_plugin_private;
	PGOutputTxnData *fc_txndata = (PGOutputTxnData *) fc_txn->output_plugin_private;
	MemoryContext fc_old;
	RelationSyncEntry *fc_relentry;
	TransactionId fc_xid = InvalidTransactionId;
	Relation	fc_ancestor = NULL;
	Relation	fc_targetrel = fc_relation;
	ReorderBufferChangeType fc_action = fc_change->action;
	TupleTableSlot *fc_old_slot = NULL;
	TupleTableSlot *fc_new_slot = NULL;

	fc_update_replication_progress(fc_ctx, false);

	if (!is_publishable_relation(fc_relation))
		return;

	/*
	 * 记住流模式下更改的xid。我们需要在流模式下每次更改时发送xid，
	 * 以便订阅者可以进行关联，并且在中止时可以丢弃相应的
	 * 更改。
	 */
	if (in_streaming)
		fc_xid = fc_change->txn->xid;

	fc_relentry = fc_get_rel_sync_entry(fc_data, fc_relation);

	/* 首先检查表过滤器 */
	switch (fc_action)
	{
		case REORDER_BUFFER_CHANGE_INSERT:
			if (!fc_relentry->pubactions.pubinsert)
				return;
			break;
		case REORDER_BUFFER_CHANGE_UPDATE:
			if (!fc_relentry->pubactions.pubupdate)
				return;
			break;
		case REORDER_BUFFER_CHANGE_DELETE:
			if (!fc_relentry->pubactions.pubdelete)
				return;
			break;
		default:
			Assert(false);
	}

	/* 通过使用并重置我们自己的上下文来避免内存泄漏 */
	fc_old = MemoryContextSwitchTo(fc_data->context);

	/* 发送数据 */
	switch (fc_action)
	{
		case REORDER_BUFFER_CHANGE_INSERT:
			fc_new_slot = fc_relentry->new_slot;
			ExecStoreHeapTuple(&fc_change->data.tp.newtuple->tuple,
							   fc_new_slot, false);

			/* 如果通过根发布，则切换关系。 */
			if (fc_relentry->publish_as_relid != RelationGetRelid(fc_relation))
			{
				Assert(fc_relation->rd_rel->relispartition);
				fc_ancestor = RelationIdGetRelation(fc_relentry->publish_as_relid);
				fc_targetrel = fc_ancestor;
				/* 如果需要转换元组。 */
				if (fc_relentry->attrmap)
				{
					TupleDesc	fc_tupdesc = RelationGetDescr(fc_targetrel);

					fc_new_slot = execute_attr_map_slot(fc_relentry->attrmap,
													 fc_new_slot,
													 MakeTupleTableSlot(fc_tupdesc, &TTSOpsVirtual));
				}
			}

			/* 检查行过滤器 */
			if (!fc_pgoutput_row_filter(fc_targetrel, NULL, &fc_new_slot, fc_relentry,
									 &fc_action))
				break;

			/*
			 * 如果我们还没有发送，则发送BEGIN。
			 *
			 * 在确保我们实际会发送更改后发送BEGIN消息。这可以避免为空事务发送一对BEGIN/COMMIT
			 * 消息。
			 */
			if (fc_txndata && !fc_txndata->sent_begin_txn)
				fc_pgoutput_send_begin(fc_ctx, fc_txn);

			/*
			 * Schema 应该使用原始关系发送，因为它
			 * 还发送祖先的关系。
			 */
			fc_maybe_send_schema(fc_ctx, fc_change, fc_relation, fc_relentry);

			OutputPluginPrepareWrite(fc_ctx, true);
			logicalrep_write_insert(fc_ctx->out, fc_xid, fc_targetrel, fc_new_slot,
									fc_data->binary, fc_relentry->columns);
			OutputPluginWrite(fc_ctx, true);
			break;
		case REORDER_BUFFER_CHANGE_UPDATE:
			if (fc_change->data.tp.oldtuple)
			{
				fc_old_slot = fc_relentry->old_slot;
				ExecStoreHeapTuple(&fc_change->data.tp.oldtuple->tuple,
								   fc_old_slot, false);
			}

			fc_new_slot = fc_relentry->new_slot;
			ExecStoreHeapTuple(&fc_change->data.tp.newtuple->tuple,
							   fc_new_slot, false);

			/* 如果通过根发布，则切换关系。 */
			if (fc_relentry->publish_as_relid != RelationGetRelid(fc_relation))
			{
				Assert(fc_relation->rd_rel->relispartition);
				fc_ancestor = RelationIdGetRelation(fc_relentry->publish_as_relid);
				fc_targetrel = fc_ancestor;
				/* 如有需要，转换元组。 */
				if (fc_relentry->attrmap)
				{
					TupleDesc	fc_tupdesc = RelationGetDescr(fc_targetrel);

					if (fc_old_slot)
						fc_old_slot = execute_attr_map_slot(fc_relentry->attrmap,
														 fc_old_slot,
														 MakeTupleTableSlot(fc_tupdesc, &TTSOpsVirtual));

					fc_new_slot = execute_attr_map_slot(fc_relentry->attrmap,
													 fc_new_slot,
													 MakeTupleTableSlot(fc_tupdesc, &TTSOpsVirtual));
				}
			}

			/* 检查行过滤器 */
			if (!fc_pgoutput_row_filter(fc_targetrel, fc_old_slot, &fc_new_slot,
									 fc_relentry, &fc_action))
				break;

			/* 如果我们还没有，则发送 BEGIN */
			if (fc_txndata && !fc_txndata->sent_begin_txn)
				fc_pgoutput_send_begin(fc_ctx, fc_txn);

			fc_maybe_send_schema(fc_ctx, fc_change, fc_relation, fc_relentry);

			OutputPluginPrepareWrite(fc_ctx, true);

			/*
			 * 根据旧元组和新元组的行过滤器结果，
			 * 更新可以被转换为插入或删除。
			 */
			switch (fc_action)
			{
				case REORDER_BUFFER_CHANGE_INSERT:
					logicalrep_write_insert(fc_ctx->out, fc_xid, fc_targetrel,
											fc_new_slot, fc_data->binary,
											fc_relentry->columns);
					break;
				case REORDER_BUFFER_CHANGE_UPDATE:
					logicalrep_write_update(fc_ctx->out, fc_xid, fc_targetrel,
											fc_old_slot, fc_new_slot, fc_data->binary,
											fc_relentry->columns);
					break;
				case REORDER_BUFFER_CHANGE_DELETE:
					logicalrep_write_delete(fc_ctx->out, fc_xid, fc_targetrel,
											fc_old_slot, fc_data->binary,
											fc_relentry->columns);
					break;
				default:
					Assert(false);
			}

			OutputPluginWrite(fc_ctx, true);
			break;
		case REORDER_BUFFER_CHANGE_DELETE:
			if (fc_change->data.tp.oldtuple)
			{
				fc_old_slot = fc_relentry->old_slot;

				ExecStoreHeapTuple(&fc_change->data.tp.oldtuple->tuple,
								   fc_old_slot, false);

				/* 如果通过根发布，则切换关系。 */
				if (fc_relentry->publish_as_relid != RelationGetRelid(fc_relation))
				{
					Assert(fc_relation->rd_rel->relispartition);
					fc_ancestor = RelationIdGetRelation(fc_relentry->publish_as_relid);
					fc_targetrel = fc_ancestor;
					/* 如果需要转换元组。 */
					if (fc_relentry->attrmap)
					{
						TupleDesc	fc_tupdesc = RelationGetDescr(fc_targetrel);

						fc_old_slot = execute_attr_map_slot(fc_relentry->attrmap,
														 fc_old_slot,
														 MakeTupleTableSlot(fc_tupdesc, &TTSOpsVirtual));
					}
				}

				/* 检查行过滤器 */
				if (!fc_pgoutput_row_filter(fc_targetrel, fc_old_slot, &fc_new_slot,
										 fc_relentry, &fc_action))
					break;

				/* 如果我们还没有，则发送 BEGIN */
				if (fc_txndata && !fc_txndata->sent_begin_txn)
					fc_pgoutput_send_begin(fc_ctx, fc_txn);

				fc_maybe_send_schema(fc_ctx, fc_change, fc_relation, fc_relentry);

				OutputPluginPrepareWrite(fc_ctx, true);
				logicalrep_write_delete(fc_ctx->out, fc_xid, fc_targetrel,
										fc_old_slot, fc_data->binary,
										fc_relentry->columns);
				OutputPluginWrite(fc_ctx, true);
			}
			else
				elog(DEBUG1, "didn't send DELETE change because of missing oldtuple");
			break;
		default:
			Assert(false);
	}

	if (RelationIsValid(fc_ancestor))
	{
		RelationClose(fc_ancestor);
		fc_ancestor = NULL;
	}

	/* 删除用于存储转换元组的新槽。 */
	if (fc_relentry->attrmap)
	{
		if (fc_old_slot)
			ExecDropSingleTupleTableSlot(fc_old_slot);

		if (fc_new_slot)
			ExecDropSingleTupleTableSlot(fc_new_slot);
	}

	/* 清理 */
	MemoryContextSwitchTo(fc_old);
	MemoryContextReset(fc_data->context);
}

static void fc_pgoutput_truncate(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
				  int fc_nrelations, Relation fc_relations[], ReorderBufferChange *fc_change)
{
	PGOutputData *fc_data = (PGOutputData *) fc_ctx->output_plugin_private;
	PGOutputTxnData *fc_txndata = (PGOutputTxnData *) fc_txn->output_plugin_private;
	MemoryContext fc_old;
	RelationSyncEntry *fc_relentry;
	int			fc_i;
	int			fc_nrelids;
	Oid		   *fc_relids;
	TransactionId fc_xid = InvalidTransactionId;

	fc_update_replication_progress(fc_ctx, false);

	/* 在流模式中记住变更的 xid。参见 pgoutput_change。 */
	if (in_streaming)
		fc_xid = fc_change->txn->xid;

	fc_old = MemoryContextSwitchTo(fc_data->context);

	fc_relids = palloc0(fc_nrelations * sizeof(Oid));
	fc_nrelids = 0;

	for (fc_i = 0; fc_i < fc_nrelations; fc_i++)
	{
		Relation	fc_relation = fc_relations[fc_i];
		Oid			fc_relid = RelationGetRelid(fc_relation);

		if (!is_publishable_relation(fc_relation))
			continue;

		fc_relentry = fc_get_rel_sync_entry(fc_data, fc_relation);

		if (!fc_relentry->pubactions.pubtruncate)
			continue;

		/*
		 * 如果发布只想通过根表发送，
		 * 则不要发送分区。
		 */
		if (fc_relation->rd_rel->relispartition &&
			fc_relentry->publish_as_relid != fc_relid)
			continue;

		fc_relids[fc_nrelids++] = fc_relid;

		/* 如果我们还没有，则发送 BEGIN */
		if (fc_txndata && !fc_txndata->sent_begin_txn)
			fc_pgoutput_send_begin(fc_ctx, fc_txn);

		fc_maybe_send_schema(fc_ctx, fc_change, fc_relation, fc_relentry);
	}

	if (fc_nrelids > 0)
	{
		OutputPluginPrepareWrite(fc_ctx, true);
		logicalrep_write_truncate(fc_ctx->out,
								  fc_xid,
								  fc_nrelids,
								  fc_relids,
								  fc_change->data.truncate.cascade,
								  fc_change->data.truncate.restart_seqs);
		OutputPluginWrite(fc_ctx, true);
	}

	MemoryContextSwitchTo(fc_old);
	MemoryContextReset(fc_data->context);
}

static void fc_pgoutput_message(LogicalDecodingContext *fc_ctx, ReorderBufferTXN *fc_txn,
				 XLogRecPtr fc_message_lsn, bool fc_transactional, const char *fc_prefix, Size fc_sz,
				 const char *fc_message)
{
	PGOutputData *fc_data = (PGOutputData *) fc_ctx->output_plugin_private;
	TransactionId fc_xid = InvalidTransactionId;

	fc_update_replication_progress(fc_ctx, false);

	if (!fc_data->messages)
		return;

	/*
	 * 在流模式中记住消息的 xid。见
	 * pgoutput_change。
	 */
	if (in_streaming)
		fc_xid = fc_txn->xid;

	/*
	 * 如果我们还没有输出 BEGIN。避免用于非事务消息。
	 */
	if (fc_transactional)
	{
		PGOutputTxnData *fc_txndata = (PGOutputTxnData *) fc_txn->output_plugin_private;

		/* 如果我们还没有，则发送 BEGIN */
		if (fc_txndata && !fc_txndata->sent_begin_txn)
			fc_pgoutput_send_begin(fc_ctx, fc_txn);
	}

	OutputPluginPrepareWrite(fc_ctx, true);
	logicalrep_write_message(fc_ctx->out,
							 fc_xid,
							 fc_message_lsn,
							 fc_transactional,
							 fc_prefix,
							 fc_sz,
							 fc_message);
	OutputPluginWrite(fc_ctx, true);
}

/*
 * 目前我们总是进行转发。
 */
static bool fc_pgoutput_origin_filter(LogicalDecodingContext *fc_ctx,
					   RepOriginId fc_origin_id)
{
	return false;
}

/*
 * 关闭输出插件。
 *
 * 注意，我们不需要清理 data->context, data->cachectx 和 pubctx，
 * 因为它们是 ctx->context 的子上下文，
 * 所以它们将由逻辑解码机制清理。
 */
static void fc_pgoutput_shutdown(LogicalDecodingContext *fc_ctx)
{
	if (RelationSyncCache)
	{
		hash_destroy(RelationSyncCache);
		RelationSyncCache = NULL;
	}

	/* 宁可安全也不要抱歉 */
	pubctx = NULL;
}

/*
 * 从发布名称列表中加载发布。
 */
static List * fc_LoadPublications(List *fc_pubnames)
{
	List	   *fc_result = NIL;
	ListCell   *fc_lc;

	foreach(fc_lc, fc_pubnames)
	{
		char	   *fc_pubname = (char *) lfirst(fc_lc);
		Publication *fc_pub = GetPublicationByName(fc_pubname, false);

		fc_result = lappend(fc_result, fc_pub);
	}

	return fc_result;
}

/*
 * 发布 syscache 作废回调。
 *
 * 在 pg_publication 上作废时调用。
 */
static void fc_publication_invalidation_cb(Datum fc_arg, int fc_cacheid, uint32 fc_hashvalue)
{
	publications_valid = false;

	/*
	 * 还要使每个关系缓存作废，
	 * 以便下次检查过滤信息时，将使用新的发布设置更新。
	 */
	fc_rel_sync_cache_publication_cb(fc_arg, fc_cacheid, fc_hashvalue);
}

/*
 * START STREAM 回调
 */
static void fc_pgoutput_stream_start(struct LogicalDecodingContext *fc_ctx,
					  ReorderBufferTXN *fc_txn)
{
	bool		fc_send_replication_origin = fc_txn->origin_id != InvalidRepOriginId;

	/* 我们不能嵌套事务的流。 */
	Assert(!in_streaming);

	/*
	 * 如果我们已经为该事务发送了第一个流，
	 * 则在后续流中不要发送源 ID。
	 */
	if (rbtxn_is_streamed(fc_txn))
		fc_send_replication_origin = false;

	OutputPluginPrepareWrite(fc_ctx, !fc_send_replication_origin);
	logicalrep_write_stream_start(fc_ctx->out, fc_txn->xid, !rbtxn_is_streamed(fc_txn));

	fc_send_repl_origin(fc_ctx, fc_txn->origin_id, InvalidXLogRecPtr,
					 fc_send_replication_origin);

	OutputPluginWrite(fc_ctx, true);

	/* 我们现在正在流式处理一部分事务 */
	in_streaming = true;
}

/*
 * STOP STREAM 回调
 */
static void fc_pgoutput_stream_stop(struct LogicalDecodingContext *fc_ctx,
					 ReorderBufferTXN *fc_txn)
{
	/* 我们应该正在流式处理一个事务 */
	Assert(in_streaming);

	OutputPluginPrepareWrite(fc_ctx, true);
	logicalrep_write_stream_stop(fc_ctx->out);
	OutputPluginWrite(fc_ctx, true);

	/* 我们已停止流式处理事务 */
	in_streaming = false;
}

/*
 * 通知下游放弃已流式传输的事务（连同所有
 * 子事务，如果它是顶级事务）。
 */
static void fc_pgoutput_stream_abort(struct LogicalDecodingContext *fc_ctx,
					  ReorderBufferTXN *fc_txn,
					  XLogRecPtr fc_abort_lsn)
{
	ReorderBufferTXN *fc_toptxn;

	/*
	 * 中止应该发生在流块外，即使对于流式
	 * 事务。虽然该事务必须标记为已流式传输。
	 */
	Assert(!in_streaming);

	/* 确定顶级事务 */
	fc_toptxn = (fc_txn->toptxn) ? fc_txn->toptxn : fc_txn;

	Assert(rbtxn_is_streamed(fc_toptxn));

	OutputPluginPrepareWrite(fc_ctx, true);
	logicalrep_write_stream_abort(fc_ctx->out, fc_toptxn->xid, fc_txn->xid);
	OutputPluginWrite(fc_ctx, true);

	fc_cleanup_rel_sync_cache(fc_toptxn->xid, false);
}

/*
 * 通知下游应用已流式传输的事务（连同所有
 * 子事务）。
 */
static void fc_pgoutput_stream_commit(struct LogicalDecodingContext *fc_ctx,
					   ReorderBufferTXN *fc_txn,
					   XLogRecPtr fc_commit_lsn)
{
	/*
	 * 提交应该发生在流块外，即使对于流式
	 * 事务。虽然该事务必须标记为已流式传输。
	 */
	Assert(!in_streaming);
	Assert(rbtxn_is_streamed(fc_txn));

	fc_update_replication_progress(fc_ctx, false);

	OutputPluginPrepareWrite(fc_ctx, true);
	logicalrep_write_stream_commit(fc_ctx->out, fc_txn, fc_commit_lsn);
	OutputPluginWrite(fc_ctx, true);

	fc_cleanup_rel_sync_cache(fc_txn->xid, true);
}

/*
 * PREPARE 回调（用于流式两阶段提交）。
 *
 * 通知下游准备事务。
 */
static void fc_pgoutput_stream_prepare_txn(LogicalDecodingContext *fc_ctx,
							ReorderBufferTXN *fc_txn,
							XLogRecPtr fc_prepare_lsn)
{
	Assert(rbtxn_is_streamed(fc_txn));

	fc_update_replication_progress(fc_ctx, false);
	OutputPluginPrepareWrite(fc_ctx, true);
	logicalrep_write_stream_prepare(fc_ctx->out, fc_txn, fc_prepare_lsn);
	OutputPluginWrite(fc_ctx, true);
}

/*
 * 初始化解码会话的关系模式同步缓存。
 *
 * 哈希表在解码会话结束时被销毁。虽然
 * relcache 无效化仍然存在并且仍将被调用，但它们
 * 只会看到 null 哈希表全局并不采取任何行动。
 */
static void fc_init_rel_sync_cache(MemoryContext fc_cachectx)
{
	HASHCTL		fc_ctl;
	static bool fc_relation_callbacks_registered = false;

	/* 如果哈希表已经存在，则不需要做任何事情 */
	if (RelationSyncCache != NULL)
		return;

	/* 为缓存创建一个新的哈希表 */
	fc_ctl.keysize = sizeof(Oid);
	fc_ctl.entrysize = sizeof(RelationSyncEntry);
	fc_ctl.hcxt = fc_cachectx;

	RelationSyncCache = hash_create("logical replication output relation cache",
									128, &fc_ctl,
									HASH_ELEM | HASH_CONTEXT | HASH_BLOBS);

	Assert(RelationSyncCache != NULL);

	/* 如果我们已经注册了回调，则不需要再做任何事情 */
	if (fc_relation_callbacks_registered)
		return;

	CacheRegisterRelcacheCallback(fc_rel_sync_cache_relation_cb, (Datum) 0);
	CacheRegisterSyscacheCallback(PUBLICATIONRELMAP,
								  fc_rel_sync_cache_publication_cb,
								  (Datum) 0);
	CacheRegisterSyscacheCallback(PUBLICATIONNAMESPACEMAP,
								  fc_rel_sync_cache_publication_cb,
								  (Datum) 0);

	fc_relation_callbacks_registered = true;
}

/*
 * 我们预计流式事务的数量相对较少。
 */
static bool fc_get_schema_sent_in_streamed_txn(RelationSyncEntry *fc_entry, TransactionId fc_xid)
{
	ListCell   *fc_lc;

	foreach(fc_lc, fc_entry->streamed_txns)
	{
		if (fc_xid == (uint32) lfirst_int(fc_lc))
			return true;
	}

	return false;
}

/*
 * 在关系同步条目中添加已经发送过模式的 xid。
 */
static void fc_set_schema_sent_in_streamed_txn(RelationSyncEntry *fc_entry, TransactionId fc_xid)
{
	MemoryContext fc_oldctx;

	fc_oldctx = MemoryContextSwitchTo(CacheMemoryContext);

	fc_entry->streamed_txns = lappend_int(fc_entry->streamed_txns, fc_xid);

	MemoryContextSwitchTo(fc_oldctx);
}

/*
 * 在关系模式缓存中查找或创建条目。
 *
 * 这会查找给定关系直接或间接参与的出版物（如果是
 * 关系的祖先部分参与出版物，则是间接的），并用
 * 发布哪些操作以及发布时是否使用祖先的模式的信息填写找到的条目。
 */
static RelationSyncEntry *
fc_get_rel_sync_entry(PGOutputData *fc_data, Relation fc_relation)
{
	RelationSyncEntry *fc_entry;
	bool		fc_found;
	MemoryContext fc_oldctx;
	Oid			fc_relid = RelationGetRelid(fc_relation);

	Assert(RelationSyncCache != NULL);

	/* 查找缓存的关系信息，如果未找到则创建 */
	fc_entry = (RelationSyncEntry *) hash_search(RelationSyncCache,
											  (void *) &fc_relid,
											  HASH_ENTER, &fc_found);
	Assert(fc_entry != NULL);

	/* 如果是新的，则初始化条目 */
	if (!fc_found)
	{
		fc_entry->replicate_valid = false;
		fc_entry->schema_sent = false;
		fc_entry->streamed_txns = NIL;
		fc_entry->pubactions.pubinsert = fc_entry->pubactions.pubupdate =
			fc_entry->pubactions.pubdelete = fc_entry->pubactions.pubtruncate = false;
		fc_entry->new_slot = NULL;
		fc_entry->old_slot = NULL;
		memset(fc_entry->exprstate, 0, sizeof(fc_entry->exprstate));
		fc_entry->entry_cxt = NULL;
		fc_entry->publish_as_relid = InvalidOid;
		fc_entry->columns = NULL;
		fc_entry->attrmap = NULL;
	}

	/* 验证条目 */
	if (!fc_entry->replicate_valid)
	{
		Oid			fc_schemaId = get_rel_namespace(fc_relid);
		List	   *fc_pubids = GetRelationPublications(fc_relid);

		/*
		 * 我们不对命名空间系统表获取锁，因为我们使用历史快照构建
		 * 缓存条目，所有后续更改在解码 WAL 时都会被吸收。
		 */
		List	   *fc_schemaPubids = GetSchemaPublications(fc_schemaId);
		ListCell   *fc_lc;
		Oid			fc_publish_as_relid = fc_relid;
		int			fc_publish_ancestor_level = 0;
		bool		fc_am_partition = get_rel_relispartition(fc_relid);
		char		fc_relkind = get_rel_relkind(fc_relid);
		List	   *fc_rel_publications = NIL;

		/* 在使用之前重新加载出版物（如果需要）。 */
		if (!publications_valid)
		{
			Assert(pubctx);

			MemoryContextReset(pubctx);
			fc_oldctx = MemoryContextSwitchTo(pubctx);

			fc_data->publications = fc_LoadPublications(fc_data->publication_names);
			MemoryContextSwitchTo(fc_oldctx);
			publications_valid = true;
		}

		/*
		 * 重置 schema_sent 状态，因为关系定义可能已经
		 * 更改。也将 pubactions 重置为空，以防关系被从
		 * 发布物中删除。还释放任何依赖于
		 * 早期定义的对象。
		 */
		fc_entry->schema_sent = false;
		list_free(fc_entry->streamed_txns);
		fc_entry->streamed_txns = NIL;
		bms_free(fc_entry->columns);
		fc_entry->columns = NULL;
		fc_entry->pubactions.pubinsert = false;
		fc_entry->pubactions.pubupdate = false;
		fc_entry->pubactions.pubdelete = false;
		fc_entry->pubactions.pubtruncate = false;

		/*
		 * 元组槽清理。（如有必要，将在稍后重建）。
		 */
		if (fc_entry->old_slot)
		{
			TupleDesc	fc_desc = fc_entry->old_slot->tts_tupleDescriptor;

			Assert(fc_desc->tdrefcount == -1);

			ExecDropSingleTupleTableSlot(fc_entry->old_slot);

			/*
			 * ExecDropSingleTupleTableSlot() 不会释放 TupleDesc，所以
			 * 现在执行此操作以避免任何泄漏。
			 */
			FreeTupleDesc(fc_desc);
		}
		if (fc_entry->new_slot)
		{
			TupleDesc	fc_desc = fc_entry->new_slot->tts_tupleDescriptor;

			Assert(fc_desc->tdrefcount == -1);

			ExecDropSingleTupleTableSlot(fc_entry->new_slot);

			/*
			 * ExecDropSingleTupleTableSlot() 不会释放 TupleDesc，所以
			 * 现在执行此操作以避免任何泄漏。
			 */
			FreeTupleDesc(fc_desc);
		}

		fc_entry->old_slot = NULL;
		fc_entry->new_slot = NULL;

		if (fc_entry->attrmap)
			free_attrmap(fc_entry->attrmap);
		fc_entry->attrmap = NULL;

		/*
		 * 行过滤器缓存清理。
		 */
		if (fc_entry->entry_cxt)
			MemoryContextDelete(fc_entry->entry_cxt);

		fc_entry->entry_cxt = NULL;
		fc_entry->estate = NULL;
		memset(fc_entry->exprstate, 0, sizeof(fc_entry->exprstate));

		/*
		 * 构建出版物缓存。我们不能使用 relcache 提供的
		 * 因为 relcache 考虑了给定关系所在的所有出版物，
		 * 但在这里我们只需要考虑订阅者请求的那些。
		 */
		foreach(fc_lc, fc_data->publications)
		{
			Publication *fc_pub = lfirst(fc_lc);
			bool		fc_publish = false;

			/*
			 * 在这个出版物中我们应该用什么 relid 发布更改？
			 * 我们将使用所有出版物中的最上层 relid。同时
			 * 跟踪这个出版物的祖先级别。
			 */
			Oid			fc_pub_relid = fc_relid;
			int			fc_ancestor_level = 0;

			/*
			 * 如果这是一个 FOR ALL TABLES 出版物，选择分区
			 * 根并相应地设置祖先级别。
			 */
			if (fc_pub->alltables)
			{
				fc_publish = true;
				if (fc_pub->pubviaroot && fc_am_partition)
				{
					List	   *fc_ancestors = get_partition_ancestors(fc_relid);

					fc_pub_relid = llast_oid(fc_ancestors);
					fc_ancestor_level = list_length(fc_ancestors);
				}
			}

			if (!fc_publish)
			{
				bool		fc_ancestor_published = false;

				/*
				 * 对于一个分区，检查任何祖先是否被
				 * 发布。如果是，则记录通过这个出版物发布的
				 * 最上层祖先，这将用作发布分区更改的
				 * 关系。
				 */
				if (fc_am_partition)
				{
					Oid			fc_ancestor;
					int			fc_level;
					List	   *fc_ancestors = get_partition_ancestors(fc_relid);

					fc_ancestor = GetTopMostAncestorInPublication(fc_pub->oid,
															   fc_ancestors,
															   &fc_level);

					if (fc_ancestor != InvalidOid)
					{
						fc_ancestor_published = true;
						if (fc_pub->pubviaroot)
						{
							fc_pub_relid = fc_ancestor;
							fc_ancestor_level = fc_level;
						}
					}
				}

				if (list_member_oid(fc_pubids, fc_pub->oid) ||
					list_member_oid(fc_schemaPubids, fc_pub->oid) ||
					fc_ancestor_published)
					fc_publish = true;
			}

			/*
			 * 如果该关系将被发布，确定发布的操作，
			 * 并列出列，如果适用的话。
			 *
			 * 不要发布分区表的更改，因为发布其分区的更改就足够了，
			 * 除非由于设置了 pubviaroot，分区更改不会被发布。
			 */
			if (fc_publish &&
				(fc_relkind != RELKIND_PARTITIONED_TABLE || fc_pub->pubviaroot))
			{
				fc_entry->pubactions.pubinsert |= fc_pub->pubactions.pubinsert;
				fc_entry->pubactions.pubupdate |= fc_pub->pubactions.pubupdate;
				fc_entry->pubactions.pubdelete |= fc_pub->pubactions.pubdelete;
				fc_entry->pubactions.pubtruncate |= fc_pub->pubactions.pubtruncate;

				/*
				 * 我们希望将更改发布为所有出版物中的最上层祖先。
				 * 因此，我们需要检查已计算的级别是否高于新的级别。
				 * 如果是，我们可以忽略新值（因为它是子节点）。
				 * 否则，新值是一个祖先，因此我们保留它。
				 */
				if (fc_publish_ancestor_level > fc_ancestor_level)
					continue;

				/*
				 * 如果我们在树中找到了更高的祖先，则舍弃我们通过
				 * 它复制的出版物列表，使用新的祖先。
				 */
				if (fc_publish_ancestor_level < fc_ancestor_level)
				{
					fc_publish_as_relid = fc_pub_relid;
					fc_publish_ancestor_level = fc_ancestor_level;

					/* 重置此关系的出版物列表 */
					fc_rel_publications = NIL;
				}
				else
				{
					/* 相同的祖先级别，必须是相同的 OID。 */
					Assert(fc_publish_as_relid == fc_pub_relid);
				}

				/* 跟踪此祖先的出版物。 */
				fc_rel_publications = lappend(fc_rel_publications, fc_pub);
			}
		}

		fc_entry->publish_as_relid = fc_publish_as_relid;

		/*
		 * 初始化元组槽、映射和行过滤器。这些仅在发布插入、
		 * 更新或删除时使用。
		 */
		if (fc_entry->pubactions.pubinsert || fc_entry->pubactions.pubupdate ||
			fc_entry->pubactions.pubdelete)
		{
			/* 初始化元组槽和映射 */
			fc_init_tuple_slot(fc_data, fc_relation, fc_entry);

			/* 初始化行过滤器 */
			fc_pgoutput_row_filter_init(fc_data, fc_rel_publications, fc_entry);

			/* 初始化列列表 */
			fc_pgoutput_column_list_init(fc_data, fc_rel_publications, fc_entry);
		}

		list_free(fc_pubids);
		list_free(fc_schemaPubids);
		list_free(fc_rel_publications);

		fc_entry->replicate_valid = true;
	}

	return fc_entry;
}

/*
 * 清理流式事务的列表并更新 schema_sent 标志。
 *
 * 当一个流式事务提交或中止时，我们需要从模式缓存中删除
 * 顶层 XID。如果事务中止，则订阅者会简单地丢弃我们
 * 流式传输的模式记录，因此我们不需要做任何其他事情。
 *
 * 如果事务已提交，则订阅者会更新关系缓存 - 所以相应地更改
 * schema_sent 标志。
 */
static void fc_cleanup_rel_sync_cache(TransactionId fc_xid, bool fc_is_commit)
{
	HASH_SEQ_STATUS fc_hash_seq;
	RelationSyncEntry *fc_entry;
	ListCell   *fc_lc;

	Assert(RelationSyncCache != NULL);

	hash_seq_init(&fc_hash_seq, RelationSyncCache);
	while ((fc_entry = hash_seq_search(&fc_hash_seq)) != NULL)
	{
		/*
		 * 我们可以为列表中已提交的 xid 设置 schema_sent 标志，
		 * 因为这确保了订阅者将拥有相应的模式，除非该关系
		 * 有任何无效化，否则我们不需要发送它。
		 */
		foreach(fc_lc, fc_entry->streamed_txns)
		{
			if (fc_xid == (uint32) lfirst_int(fc_lc))
			{
				if (fc_is_commit)
					fc_entry->schema_sent = true;

				fc_entry->streamed_txns =
					foreach_delete_current(fc_entry->streamed_txns, fc_lc);
				break;
			}
		}
	}
}

/*
 * Relcache 无效化回调
 */
static void fc_rel_sync_cache_relation_cb(Datum fc_arg, Oid fc_relid)
{
	RelationSyncEntry *fc_entry;

	/*
	 * 如果插件在 SQL 接口中被使用，我们可以到这里，因为
	 * 当解码完成时，RelSchemaSyncCache 被销毁，但
	 * 没有办法注销 relcache 无效化回调。
	 */
	if (RelationSyncCache == NULL)
		return;

	/*
	 * 除了逻辑解码回调调用之外，没有人保留指向
	 * 此哈希表中条目的指针 - 但是如果我们在回调中
	 * 进行任何 syscache 访问，则无效化事件可以在回调期间发生。
	 * 因此，我们必须将缓存条目标记为无效，但不要损坏
	 * 其任何子结构。下一个 get_rel_sync_entry() 调用将
	 * 重建全部内容。
	 */
	if (OidIsValid(fc_relid))
	{
		/*
		 * 获取不在表中的关系的无效化完全正常。
		 * 所以我们不在乎它是否被找到。
		 */
		fc_entry = (RelationSyncEntry *) hash_search(RelationSyncCache, &fc_relid,
												  HASH_FIND, NULL);
		if (fc_entry != NULL)
			fc_entry->replicate_valid = false;
	}
	else
	{
		/* 整个缓存必须被刷新。 */
		HASH_SEQ_STATUS fc_status;

		hash_seq_init(&fc_status, RelationSyncCache);
		while ((fc_entry = (RelationSyncEntry *) hash_seq_search(&fc_status)) != NULL)
		{
			fc_entry->replicate_valid = false;
		}
	}
}

/*
 * 出版关系/模式映射 syscache 无效化回调
 *
 * 在 pg_publication、pg_publication_rel 和
 * pg_publication_namespace 上的无效化时调用。
 */
static void fc_rel_sync_cache_publication_cb(Datum fc_arg, int fc_cacheid, uint32 fc_hashvalue)
{
	HASH_SEQ_STATUS fc_status;
	RelationSyncEntry *fc_entry;

	/*
	 * 如果插件在 SQL 接口中被使用，我们可以到这里，因为
	 * 当解码完成时，RelSchemaSyncCache 被销毁，但
	 * 没有办法注销 relcache 无效化回调。
	 */
	if (RelationSyncCache == NULL)
		return;

	/*
	 * 没有办法找到哈希属于我们缓存中的哪个条目，所以
	 * 将整个缓存标记为无效。
	 */
	hash_seq_init(&fc_status, RelationSyncCache);
	while ((fc_entry = (RelationSyncEntry *) hash_seq_search(&fc_status)) != NULL)
	{
		fc_entry->replicate_valid = false;
	}
}

/* 发送复制源 */
static void fc_send_repl_origin(LogicalDecodingContext *fc_ctx, RepOriginId fc_origin_id,
				 XLogRecPtr fc_origin_lsn, bool fc_send_origin)
{
	if (fc_send_origin)
	{
		char	   *fc_origin;

		/*----------
		 * XXX：我们想要什么行为？
		 *
		 * 备选方案：
		 *  - 如果找不到源名称，则不发送源消息
		 *    （这就是我们现在做的）
		 *  - 抛出错误 - 这将破坏复制，不好
		 *  - 发送某个特殊的“未知”源
		 *----------
		 */
		if (replorigin_by_oid(fc_origin_id, true, &fc_origin))
		{
			/* 消息边界 */
			OutputPluginWrite(fc_ctx, false);
			OutputPluginPrepareWrite(fc_ctx, true);

			logicalrep_write_origin(fc_ctx->out, fc_origin, fc_origin_lsn);
		}
	}
}

/*
 * 尝试更新进度，并在处理了太多更改时发送保持活动消息。
 *
 * 对于一个大事务，如果我们在很长时间内没有向下游发送任何更改
 * （超过备用 wal_receiver_timeout），那么它可能会超时。
 * 当所有或大多数更改未被发布或被过滤掉时，这种情况可能会发生。
 */
static void fc_update_replication_progress(LogicalDecodingContext *fc_ctx, bool fc_skipped_xact)
{
	static int	fc_changes_count = 0;

	/*
	 * 我们不想在处理每个更改后尝试发送保持活动消息，因为这可能会带来开销。
	 * 测试显示，在连续处理 100 个左右更改后
	 * 这样做没有明显的开销。
	 */
#define CHANGES_THRESHOLD 100

	/*
	 * 如果我们处于事务 LSN 的末尾，请更新进度跟踪。
	 * 否则，在连续处理 CHANGES_THRESHOLD 更改后，
	 * 我们会在需要时尝试发送保持活动消息。
	 */
	if (fc_ctx->end_xact || ++fc_changes_count >= CHANGES_THRESHOLD)
	{
		OutputPluginUpdateProgress(fc_ctx, fc_skipped_xact);
		fc_changes_count = 0;
	}
}
